package com.talang.surfing.batch.controller;

import com.talang.surfing.batch.core.HasId;
import com.talang.surfing.batch.core.TaskCallback;
import com.talang.surfing.batch.core.TaskProcessor;
import com.talang.surfing.batch.dto.TaskStat;
import com.talang.surfing.batch.executor.AsyncExecutor;
import com.talang.surfing.batch.mapper.BatchMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * BaseBatchController 是一个抽象基类，主要用于批量处理一些具有类似属性的任务（例如，所有的任务都有一个 id）。
 * 该类提供了一个通用的处理模式，子类只需要根据具体的业务需求实现相应的方法即可。
 *
 * @param <T> 表示任务类型，该类型必须实现 HasId 接口
 * @param <M> 表示任务数据访问接口类型，该类型必须是 BatchMapper 的子接口
 * @author: wangwanbao,QQ: 51198641
 * @Date: 2023/06/18 7:35 PM
 */

@Slf4j
public abstract class BaseBatchController<T extends HasId, M extends BatchMapper<T>> {

    /**
     * 异步执行器，用于异步执行任务
     */
    @Resource
    private AsyncExecutor executor;

    /**
     * 异步执行器，用于异步执行任务
     */
    @Resource
    private BatchMapper mapper;

    /**
     * 每次处理的任务数，可按需要调整
     */
    private Integer pageSize = 2;

    /**
     * 上次处理的任务的最大值，通常是ID，某些特殊情况下表没有ID，或者联表查询，可以自定义这个值，
     * resetLastValue用于计算下一次任务的起始值
     */
    private Object lastValue = 0;

    /**
     * 成功处理的任务数
     */
    private AtomicInteger success = new AtomicInteger(0);

    /**
     * 成功处理的任务数
     */
    private AtomicInteger fail = new AtomicInteger(0);

    /**
     * 该方法实现了一种批量处理任务的通用模式，它会循环获取并处理任务，直到所有任务都处理完毕。
     *
     * @param taskName      任务名称，用于日志记录
     * @param taskProcessor 任务处理器，用于处理具体的任务
     * @param taskCallback  任务回调接口，用于处理任务执行结果
     * @return 任务处理统计信息
     */
    public TaskStat process(String taskName, TaskProcessor<T> taskProcessor, TaskCallback<T> taskCallback) {
        int total = mapper.selectCount();
        while (true) {
            AtomicInteger currentSuccess = new AtomicInteger(0);
            AtomicInteger currentFail = new AtomicInteger(0);
            List<T> tasks = mapper.fetchTasks(lastValue, pageSize);
            if (CollectionUtils.isEmpty(tasks)) {
                log.info("[{}] 任务执行完毕", taskName);
                break;
            }
            resetLastValue(tasks);
            List<Future<Boolean>> futures = tasks.stream().map(task -> executor.execute(task, taskProcessor, taskCallback))
                    .collect(Collectors.toList());

            futures.forEach(future -> {
                try {
                    if (future.get()) {
                        currentSuccess.incrementAndGet();
                        success.incrementAndGet();
                    } else {
                        currentFail.incrementAndGet();
                        fail.incrementAndGet();
                    }
                } catch (Exception e) {
                    currentFail.incrementAndGet();
                    fail.incrementAndGet();
                    e.printStackTrace();
                }
            });
            log.info("任务{}, 本次执行{}个任务，成功{}个，失败{}个", taskName, tasks.size(), currentSuccess.get(), currentFail.get());
        }
        return TaskStat.builder()
                .taskName(taskName)
                .total(total)
                .success(success.get())
                .fail(fail.get())
                .build();
    }

    /**
     * 该方法用于更新 lastValue，即获取当前任务列表中 id 最大的任务的 id。
     *
     * @param tasks 当前任务列表
     */
    protected void resetLastValue(List<T> tasks) {
        if (tasks == null || tasks.isEmpty()) {
            return;
        }
        T maxIdTask = tasks.stream()
                .max(Comparator.comparing(task -> task.getId()))
                .orElse(null);
        if (maxIdTask != null) {
            this.lastValue = maxIdTask.getId();
        }
    }

    /**
     * 抽象方法，需要子类根据具体的业务逻辑来实现。通常情况下，子类会在这个方法中调用 process 方法来处理任务。
     *
     * @return 任务处理统计信息
     */
    protected abstract TaskStat doBatch();


}
